package com.example.source;

/**
 *
 *  * @projectName myflinkstu
 *  * @title     ClickSource
 *  * @package    com.example.source
 *  * @description
 *  * @author hjx
 *  * @date   2022-3-28 10:03
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *
 */

import com.example.beans.Event;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;
public class ClickSource implements SourceFunction<Event> {
    // 声明一个布尔变量，作为控制数据生成的标识位
    private Boolean running = true;
    private Long sleepTime = 1000L;

    public ClickSource() {
    }

    public ClickSource(Long sleepTime) {
        this.sleepTime = sleepTime;
    }

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        Random random = new Random(); // 在指定的数据集中随机选取数据
        String[] users = {"Mary", "Alice", "Bob", "Cary","Amy","Tony","Tom","Jerry","Orson","Ken"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2","./pay","./non-pay","./exit"};
        while (running) {
            ctx.collect(new Event(
                    users[random.nextInt(users.length)],
                    urls[random.nextInt(urls.length)],
                    Calendar.getInstance().getTimeInMillis()
            ));
            // 隔 1 秒生成一个点击事件，方便观测
            Thread.sleep(sleepTime);
        }
    }
    @Override
    public void cancel() {
        running = false;
    }
}

